package com.udf.flink.scala.apitest.window

import java.lang
import java.text.SimpleDateFormat
import java.time.Duration

import com.udf.flink.scala.apitest.checkpoint.Obj1
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, Watermark, WatermarkGenerator, WatermarkGeneratorSupplier, WatermarkOutput, WatermarkStrategy}
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.scala.function.WindowFunction
//import org.apache.flink.streaming.api.functions.windowing.WindowFunction
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
//import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
//import org.apache.flink.table.runtime.generated.WatermarkGenerator
import org.apache.flink.util.Collector
import org.apache.flink.api.scala._

class MinDataReduceFunction extends ReduceFunction[Obj1]{
  override def reduce(t1: Obj1, t2: Obj1): Obj1 = {
    if (t1.time > t2.time) {
      t1
    } else {
      t2
    }
  }
}


class PrintWindElenemt extends ProcessWindowFunction[Obj1,String,String,TimeWindow]{
  override def process(key: String, context: Context, elements: Iterable[Obj1], out: Collector[String]): Unit = {
    //    StringBuilder allRecords= new StringBuilder();
    //    for (Obj1 element : elements) {
    //      allRecords.append(element).append("\n");
    //    }
    //
    //    String windowStart=new DateTime(context.window().getStart(), DateTimeZone.forID("+08:00")).toString("yyyy-MM-dd HH:mm:ss");
    //    String windowEnd=new DateTime(context.window().getEnd(), DateTimeZone.forID("+08:00")).toString("yyyy-MM-dd HH:mm:ss");
    //
    //    String result="Key: "+key+" 窗口开始时间: "+windowStart+" 窗口结束时间: "+windowEnd+" 窗口所有数据: \n"+allRecords;
    //    out.collect(result);
    var count = 0
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    var strs = ""
    val wins=
      s"""window key：${key},开始时间：${sdf.format(context.window.getStart)},结束时间：${sdf.format(context.window.getEnd)},maxTime：${sdf.format(context.window.maxTimestamp())}""".stripMargin

    // 遍历，获得窗口所有数据
    for (obj <- elements) {
//      println(obj.toString)
      strs=strs+"\n"+obj.toString
      count += 1
    }
    out.collect(wins+s" 数据量 : ${count}"+strs)

  }
}

class PrintWindElenemtApply extends WindowFunction[Obj1,String,String,TimeWindow]{
  override def apply(key: String, window: TimeWindow, input: Iterable[Obj1], out: Collector[String]): Unit = {


      var count = 0
      val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
      var strs = ""
      val wins=
        s"""window key：${key},开始时间：${sdf.format(window.getStart)},结束时间：${sdf.format(window.getEnd)},maxTime：${sdf.format(window.maxTimestamp())}""".stripMargin
//      val ite =input.iterator()
      // 遍历，获得窗口所有数据
    for (obj <- input) {
//      val tup2 = ite.next()
//      arrBuff.append(tup2._2)
      strs=strs+"\n"+obj.toString
             count += 1
    }

      out.collect(wins+s" 数据量 : ${count}"+strs)


  }
}

class MyWaterMarker extends SerializableTimestampAssigner[Obj1]
{
  var currentTimeMillis:Long = 0L
  var timeDiff:Long = 10000L
  var sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")

//  def getCurrentWatermark: Watermark = {
//    val watermark: Watermark = new Watermark(currentTimeMillis - timeDiff)
//    watermark
//  }

  override def extractTimestamp(element: Obj1, previousElementTimestamp: Long): Long = {
    val eventTime = element.time * 1000
    currentTimeMillis = Math.max(eventTime, currentTimeMillis)
    val id = Thread.currentThread().getId

//    println(
//      "currentThreadId:" + id + ",key:" + element.id
//        + ",eventTime:[" + element.time* 1000 + "|" + sdf.format(element.time* 1000) + "]"
//        + ",currentMaxTimeStamp:[" + currentTimeMillis + "|" + sdf.format(currentTimeMillis) + "]"
//        + ",waterMark:[" + this.getCurrentWatermark + "|" + sdf.format(this.getCurrentWatermark.getTimestamp) + "]"
//    )

    eventTime
  }
}

class BoundedOutOfOrdernessGenerator extends WatermarkGenerator[Obj1] {

  var maxOutOfOrderness :Long =3500 // 3.5 秒

  var currentMaxTimestamp: Long = _
  def BoundedOutOfOrdernessWatermarks(OutOfOrderness: Duration) {
//    checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
//    checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");

    maxOutOfOrderness = OutOfOrderness.toMillis;

    // start so that our lowest watermark would be Long.MIN_VALUE.
    currentMaxTimestamp = Long.MinValue + maxOutOfOrderness + 1;
  }


  override def onEvent(element: Obj1, eventTimestamp: Long,output: WatermarkOutput): Unit = {
    currentMaxTimestamp = Math.max(eventTimestamp, currentMaxTimestamp)
  }

  override def onPeriodicEmit(output: WatermarkOutput ): Unit = {
    // 发出的 watermark = 当前最大时间戳 - 最大乱序时间

    output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness - 1));
  }
}
//class MyWatermarkStrategy extends WatermarkStrategy[Obj1]{
//  override def createWatermarkGenerator(context: WatermarkGeneratorSupplier.Context): WatermarkGenerator[Obj1] = {
//
//
//  }
//}